package mr.diff;

import java.io.IOException;
import java.util.StringTokenizer;

import mr.CMapper;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import util.DiffComputation;
import util.DiffMessage;
import util.Parsable;

public class DifStep1<T0 extends Parsable<T0>, T1 extends Parsable<T1>, T2 extends Parsable<T2>>
		extends CMapper<T0, T1, T2, DiffMessage<T1>> {

	private static final Log LOG = LogFactory.getLog(DifStep1.class);
	// protected static final String DELIM = " ";

	private T0 cur;
	protected int flg;

	protected T1 otimsAccPrew;

	private OutputCollector<LongWritable, Text> output;

	public DifStep1(DiffComputation<T0, T1, T2> di) {
		super(di);
		flg = -1;
		chunkey = -1;
	}

	@Override
	public void close() throws IOException {
		if (chunkey >= 0) {

			System.out.println("otimsAcc: " + otimsAcc);

			// All to all
			DiffMessage<T1> msg = new DiffMessage<T1>(chunkey, otimsAcc);
			//addMesg(msg);

			// for test
			output.collect(new LongWritable(0) ,new Text(msg.toString()) );

			LOG.info("send msg:" + msg);
  
		//	System.out.println("send msg: "
		//			+ new DiffMessage<T1>(chunkey, otimsAcc).toString());

		}
	}

	@Override
	public void configure(JobConf job) {
		otimsAcc = otimes.id();

		// for test
		//job.getInputFormat().
		try {
			InputSplit[] splts = job.getInputFormat().getSplits(job, 1);
			int n = splts.length;
			String tmp = "";
			for(InputSplit spl : splts){
				tmp += spl.toString() + "[-]";
			}
			LOG.info("++++++++++++++++++CONFIG: InputSplit" + n);
			LOG.info("++++++++++++++++++CONFIG: InputSplit: " + tmp);
			
		} catch (IOException e) {
			e.printStackTrace();
		}
		
	//	System.out.println("++++++++++++configure === DifStep1");
	}

	@Override
	// KeyValueTextInputFormat
	public void map(LongWritable key, Text line,
			OutputCollector<LongWritable, Text> output, Reporter reporter)
			throws IOException {

		
		
		//System.out.println(" ==== map=>  key:" + key + ", value:" + line);
		if (chunkey < 0) {
			LOG.info("+++++++====+++++++CONFIG: map InputSplit" + reporter.getInputSplit());
			chunkey = this.getChunkey(reporter);
			flg++;

			// for test
			this.output = output;

		}

		StringTokenizer itr = new StringTokenizer(line.toString().trim(),
				ELE_Delim);

		while (itr.hasMoreTokens()) {

			String tmp = itr.nextToken();
			T0 ins = p.getInstanceOfT0();
			cur = ins.parse(tmp);
		//	System.out.println("hasMoreTokens +++++ current: " + cur);

			if (flg == 0) { // first chunk , 1st step
				if (chunkey == 0) {
					otimsAcc = cons_c;// otimes.evaluate(cons_c,
										// q.evaluate(cur));
			//		System.out.println("init  +++++ otimsAcc: " + otimsAcc);
				}
				flg++;
			}
			otimsAccPrew = otimsAcc;
			//System.out.println("hasMoreTokens1 +++++ otimsAcc: " + otimsAcc);
			otimsAcc = otimes.evaluate(otimsAcc, q.evaluate(cur));
		//	System.out.println("hasMoreTokens2 +++++ otimsAcc: " + otimsAcc);
		}

		reporter.progress();
		//System.out.println("map +++++ otimsAcc: " + otimsAcc);
	}

}